package queue

import (
	"encoding/json"
	"fmt"
	"time"

	"github.com/sirupsen/logrus"
	"github.com/spf13/viper"
    
    "collecter/app/common/enum"
    "collecter/app/common/helper"
    "collecter/app/queue/driver"
    "collecter/app/queue/interfaces"
)

var configer *viper.Viper
var logger *logrus.Logger

// 队列服务提供者映射
var providerMap = map[string]interfaces.DriverIntefacer[driver.MessageStruct]{
	"redis": driver.RedisTaskStruct{},
}

var defaultDriver string

func InitQueue() {
	fmt.Println("队列服务初始化 ... ")

	configer = helper.Configer
	logger = helper.Logger

	defaultDriver = configer.GetString("queue.defaultDriver")
	if len(defaultDriver) == 0 {
		logmsg := "初始化队列服务失败：队列驱动未配置"
		panic(logmsg)
	}
	go func() {
		queueNames := enum.GetQueueNames()
		// fmt.Println("==== mqHandler::queueNames: ", queueNames)
		for {
			// 协程中处理队列消息
			var message *driver.MessageStruct
			if msg, err := Pull(queueNames, 10); err != nil {
				errMsg := "从队列拉取消息出错，errMsg：" + err.Error()
				logger.WithField("apart", "queue").Errorln(errMsg)
				return
			} else {
				// fmt.Println(msg)
				message = msg
			}
			// fmt.Printf("===== message: %+v, handler: %+v\n", message, queueRegisters)
			if message != nil {
				go func(message *driver.MessageStruct, registers map[string]interfaces.QJInterfacer) {
					register, ok := registers[message.TaskName]
					if !ok {
						errMsg := "队列任务未配置：" + message.TaskName
						logger.WithField("apaart", "queue").Infoln(errMsg)
						handleFailed(message)
						return
					}
					execInstance := register.SetValue(message.Params)
					msgJson, _ := json.Marshal(message)
					if err := execInstance.Handle(); err != nil {
						errMsg := " 处理队列消息出错，errMsg：" + err.Error() + "消息体：" + string(msgJson)
						logger.WithField("apart", "queue").Errorln(errMsg)
						handleFailed(message)
					} else {
						handleSuccess(message)
					}
					register = nil
					execInstance = nil
					return
				}(message, queueRegisters)
			}
			time.Sleep(time.Second * 5)
		}
	}()
}

func handleSuccess(message *driver.MessageStruct) {
	msgJson, _ := json.Marshal(message)
	errMsg := "队列已执行完成"
	if err := Confirm(message.TaskName, message.Params, message.RetryTimes, message.RetryedTimes, "success"); err != nil {
		errMsg += "但发送确认消息出错。" + "message: " + string(msgJson)
	} else {
		errMsg += "。message: " + string(msgJson)
	}
	logger.WithField("apart", "queue").Infoln(errMsg)
}

func handleFailed(message *driver.MessageStruct) {
	// 消息处理失败，重新入队列
	fmt.Printf("队列消息处理失败，重新入队列。info：%+v\n", *message)

	if message.RetryTimes > 0 {
		if message.RetryedTimes <= message.RetryTimes {
			message.Key = enum.QUEUE_RETRY
			message.State = "failed"
		} else if message.RetryedTimes >= message.RetryTimes {
			// 重试次数用完，放入死信队列
			message.Key = enum.QUEUE_DEAD_LETTER
			message.State = "deadLetter"
		}
	} else {
		message.Key = enum.QUEUE_RETRY
		message.State = "failed"
	}
	if err := Confirm(message.TaskName, message.Params, message.RetryTimes, message.RetryedTimes, message.State); err != nil {
		errMsg := "队列消息处理失败，重新入队列出错：" + err.Error()
		logger.WithField("apart", "queue").Errorln(errMsg)
	}
}
